package cn.dmp.tags

import cn.dmp.utils.{JedisPools, TagsUtils}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import redis.clients.jedis.Jedis

import scala.collection.mutable.ListBuffer

object Tags4CtxV2 {
  def main(args: Array[String]): Unit = {

    if (args.length != 4) {
      println(
        """
          |cn.xiao.report.AppAnaylyseRptV2
          |参数：
          |输入路径
          |字典文件路径
          |停用字段文件路径
          |输出路径
        """.stripMargin)
      sys.exit()
    }

    val Array(logInputPath, dictFilePath, stopWordPath, outputPath) = args

    val conf: SparkConf = new SparkConf()
      .setAppName(s"${this.getClass.getSimpleName}")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)

    val hadoopConfiguration: Configuration = sc.hadoopConfiguration
    val fs: FileSystem = FileSystem.get(hadoopConfiguration)
    val newPath = new Path(outputPath)
    if (fs.exists(newPath)) {
      fs.delete(newPath, true)
    }

    val qLContext = new SQLContext(sc)

    //字典文件
    val dicMap: Map[String, String] = sc.textFile(dictFilePath).map(line => {
      val fields: Array[String] = line.split("\t", -1)
      (fields(4), fields(1))
    }).collect().toMap

    //停用字段文件
    val stopWordMap: Map[String, Int] = sc.textFile(stopWordPath).map((_, 0)).collect().toMap

    //广播变量 ---> executor
    val broadcastDic: Broadcast[Map[String, String]] = sc.broadcast(dicMap)
    val broadcastStopWord: Broadcast[Map[String, Int]] = sc.broadcast(stopWordMap)

    // 读取日志parquet文件
    qLContext.read.parquet(logInputPath)
      .where(TagsUtils.hasSomeUserIdConditition)
      .mapPartitions(par => {
        val jedis: Jedis = JedisPools.getJedis()
        val resut: Iterator[(String, List[(String, Int)])] = par.map(row => {
          //进行标签化处理
          val ads: Map[String, Int] = Tags4Ads.makeTags(row)
          val apps: Map[String, Int] = Tags4App.makeTags(row, broadcastDic.value)
          val devices: Map[String, Int] = Tags4Devices.makeTags(row)
          val keywords: Map[String, Int] = Tags4KeyWords.makeTags(row, broadcastStopWord.value)
          val business: Map[String, Int] = Tags4Buiness.makeTags(row, jedis)

          val allUserId: ListBuffer[String] = TagsUtils.getAllUserId(row)

          val otherUserId: Map[String, Int] = allUserId.slice(1,allUserId.length).map(uId =>(uId,0)).toMap



          (allUserId(0), (ads ++ apps ++ devices ++ keywords ++ business++otherUserId).toList)
        })
        jedis.close()
        resut

      }).reduceByKey((a, b) => {
      // (a ++ b).groupBy(_._1).mapValues(_.foldLeft(0)(_+_._2)).toList
      (a ++ b).groupBy(_._1).map {
        case (k, sameTags) => (k, sameTags.map(_._2).sum)
      }.toList
    }).saveAsTextFile(outputPath)

    sc.stop()
  }
}
